package com.raylu.day03basic;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

public class Example8RichParallelSourceFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .addSource(new ParaIntSource()).setParallelism(2)
                .rescale()
                .print("rescale: ").setParallelism(4);

        env
                .addSource(new ParaIntSource()).setParallelism(2)
                .rebalance()
                .print("rebalance: ").setParallelism(4);

        env.execute();
    }

    public static class ParaIntSource extends RichParallelSourceFunction<Integer> {
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println("并行子任务的索引是：" + getRuntimeContext().getIndexOfThisSubtask() + "，生命周期开始");
        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println("并行子任务的索引是：" + getRuntimeContext().getIndexOfThisSubtask() + "，生命周期结束");
        }

        @Override
        public void run(SourceContext<Integer> ctx) throws Exception {
            for (int i = 1; i < 9; i++) {
                // 如果当前的并行子任务的索引是1,那么将奇数（1,3,5,7）发送出去
                if (i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                    ctx.collect(i);
                }
            }
        }

        @Override
        public void cancel() {

        }
    }
}
